package com.shujia.flink.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo11FileSInk {
  def main(args: Array[String]): Unit = {

    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()
    /**
     * flink sql环境
     *
     */
    val table: TableEnvironment = TableEnvironment.create(settings)


    table.executeSql(
      """
        |
        |CREATE TABLE datagen (
        |    id STRING,
        |    name STRING,
        |    age INT,
        |    gender STRING,
        |    clazz STRING,
        |    ts AS localtimestamp
        |) WITH (
        | 'connector' = 'datagen',
        | 'rows-per-second' = '500', -- 每秒生成的数据行数据
        | 'fields.id.length' = '5', --字段长度限制
        | 'fields.name.length'='3',
        | 'fields.gender.length'='1',
        | 'fields.clazz.length'='5',
        | 'fields.age.min' ='1', -- 最小值
        | 'fields.age.max'='100' -- 最大值
        |)
        |
        |""".stripMargin)

    table.executeSql(
      """
        |
        |CREATE TABLE file_sink (
        |    id STRING,
        |    name STRING,
        |    age INT,
        |    gender STRING,
        |    clazz STRING,
        |    `day` STRING,
        |    `hour` STRING
        |) PARTITIONED BY (`day`,`hour`) WITH (
        |  'connector'='filesystem',
        |  'path'='data/flink_sink',
        |  'format'='csv',
        |  'sink.rolling-policy.file-size' ='100kb'
        |)
        |
        |""".stripMargin)


    table.executeSql(
      """
        |
        |insert into file_sink
        |select
        |id,name,age,gender,clazz,
        |DATE_FORMAT(ts, 'yyyy-MM-dd') as `day`,
        |DATE_FORMAT(ts, 'HH')  as `hour`
        |from
        |datagen
        |
        |""".stripMargin)
  }

}
